package com.partsoft.dits.ldap;



import java.util.Arrays;
import java.util.Collections;
import java.util.HashMap;
import java.util.HashSet;
import java.util.List;
import java.util.Map;
import java.util.Properties;
import java.util.Set;

import org.springframework.context.ApplicationContextAware;
import org.springframework.context.ApplicationEvent;
import org.springframework.context.ApplicationListener;
import org.springframework.ldap.core.ContextSource;
import org.springframework.util.Assert;
import org.springframework.util.ClassUtils;
import org.springframework.util.CollectionUtils;
import org.springframework.util.ObjectUtils;

import com.partsoft.dits.AbstractConfigurableDescriptorManager;
import com.partsoft.dits.BaseDataIntegrateBatch;
import com.partsoft.dits.DataBroadcaster;
import com.partsoft.dits.DataEdgeDescriptor;
import com.partsoft.dits.DataIntegrateBatch;
import com.partsoft.dits.DataIntegrateProcessor;
import com.partsoft.dits.DataReceiver;
import com.partsoft.dits.DataTransformer;
import com.partsoft.dits.DescriptorDepends;
import com.partsoft.dits.EdgeDescriptor;
import com.partsoft.dits.EdgeException;
import com.partsoft.dits.LdapDataEdgeDescriptor;
import com.partsoft.dits.MonitorStartMode;
import com.partsoft.dits.integrate.LdapDataDescriptorRegistry;
import com.partsoft.dits.log.ChangeLogResolver;
import com.partsoft.dits.log.ChangeLogStorage;
import com.partsoft.dits.log.Log;
import com.partsoft.dits.message.DataAction;
import com.partsoft.dits.message.DataProperty;
import com.partsoft.dits.message.MessageEntity;
import com.partsoft.dits.message.MessageFormat;
import com.partsoft.dits.message.MessageProcessor;

@SuppressWarnings({ "unchecked", "rawtypes" })
public class ContextConfigurableLdapChangeLogMonitorReactor extends
		AbstractConfigurableDescriptorManager<LdapChangeLogMonitorDescriptor> implements LdapChangeLogMonitorManager,
		DataIntegrateProcessor, ApplicationListener<ApplicationEvent>, ApplicationContextAware,
		DescriptorDepends<LdapChangeLogMonitorDescriptor> {

	private Set<Class<? extends EdgeDescriptor>> dependsClasses;
	{
		dependsClasses = new HashSet<Class<? extends EdgeDescriptor>>(1);
		dependsClasses.add(LdapContextDescriptor.class);
	}

	private Map<String, LdapChangeLogMonitorTask> monitors = Collections
			.synchronizedMap(new HashMap<String, LdapChangeLogMonitorTask>(1));

	private LdapContextSourceProvider ldapContextSourceProvider;

	private MessageProcessor messageProcessor = MessageProcessor.INSTANCE;

	private LdapDataDescriptorRegistry<LdapDataEdgeDescriptor> dataDescriptorRegistry;

	private DataTransformer dataTransformer;

	private DataBroadcaster dataBroadcaster;
	
	private DataReceiver dataReceiver;

	private ChangeLogResolver changeLogResolver;
	
	private ChangeLogStorage changeLogStorage;
	
	private String messageByteFormat = "utf-8"; 

	private String namespace;

	public Class<LdapChangeLogMonitorDescriptor> getSupportClass() {
		return LdapChangeLogMonitorDescriptor.class;
	}

	public DataBroadcaster getDataBroadcaster() {
		return dataBroadcaster;
	}
	
	public String getMessageByteFormat() {
		return messageByteFormat;
	}

	public void setMessageByteFormat(String messageByteFormat) {
		this.messageByteFormat = messageByteFormat;
	}

	public DataReceiver getDataReceiver() {
		return dataReceiver;
	}

	public void setDataBroadcaster(DataBroadcaster dataBroadcaster) {
		assertNoInit();
		this.dataBroadcaster = dataBroadcaster;
	}
	
	public void setDataReceiver(DataReceiver dataReceiver) {
		assertNoInit();
		this.dataReceiver = dataReceiver;
	}

	public void setChangeLogResolver(ChangeLogResolver changeLogResolver) {
		assertNoInit();
		this.changeLogResolver = changeLogResolver;
	}
	
	public void setChangeLogStorage(ChangeLogStorage changeLogStorage) {
		assertNoInit();
		this.changeLogStorage = changeLogStorage;
	}

	public void setLdapContextSourceProvider(LdapContextSourceProvider ldapContextSourceProvider) {
		assertNoInit();
		this.ldapContextSourceProvider = ldapContextSourceProvider;
	}

	public DataTransformer getDataTransformer() {
		return dataTransformer;
	}

	public void setDataTransformer(DataTransformer dataTransformer) {
		assertNoInit();
		this.dataTransformer = dataTransformer;
	}

	public LdapContextSourceProvider getLdapContextSourceProvider() {
		return ldapContextSourceProvider;
	}

	public String getNamespace() {
		return namespace;
	}

	public void setNamespace(String namespace) {
		assertNoInit();
		this.namespace = namespace;
	}

	public void setDataDescriptorRegistry(LdapDataDescriptorRegistry<LdapDataEdgeDescriptor> dataDescriptorRegsitry) {
		assertNoInit();
		this.dataDescriptorRegistry = dataDescriptorRegsitry;
	}

	public LdapDataDescriptorRegistry<LdapDataEdgeDescriptor> getDataDescriptorRegistry() {
		return dataDescriptorRegistry;
	}

	public void setMessageProcessor(MessageProcessor messageProcessor) {
		assertNoInit();
		this.messageProcessor = messageProcessor;
	}

	public MessageProcessor getMessageProcessor() {
		return messageProcessor;
	}

	private void createAndStartLdapChangeLogMonitor(LdapChangeLogMonitorDescriptor descriptor) {
		synchronized (this) {
			if (descriptor.getStartMode() == MonitorStartMode.DISABLED)
				return;
			ContextSource ds = getLdapContextSourceProvider().getLdapContextSource(descriptor.getLdapContext());
			LdapChangeLogMonitorContext ctx = new LdapChangeLogMonitorContextImpl(getNamespace(), ds,
					changeLogResolver, changeLogStorage, descriptor, getDataDescriptorRegistry(), this);
			LdapChangeLogMonitorTask worker = new LdapChangeLogMonitorTask(descriptor, ctx);
			monitors.put(descriptor.getLdapContext(), worker);
			Thread thread = new Thread(worker, String.format("LdapMonitor[%s]", descriptor.getLdapContext()));
			thread.start();
		}
	}

	private void stopAndRemoveLdapChangeLogMonitor(LdapChangeLogMonitorDescriptor descriptor) {
		synchronized (this) {
			LdapChangeLogMonitorTask worker = monitors.remove(descriptor.getLdapContext());
			if (worker != null)
				worker.stop();
		}
	}

	@Override
	protected void beforeRemoveDescriptor(LdapChangeLogMonitorDescriptor descriptor) {
		stopAndRemoveLdapChangeLogMonitor(descriptor);
	}

	@Override
	protected void postRemoveDescriptor(LdapChangeLogMonitorDescriptor descriptor) {
	}

	@Override
	protected void beforeLoadConfiguredDescriptor(LdapChangeLogMonitorDescriptor descriptor) {
		createAndStartLdapChangeLogMonitor(descriptor);
	}

	@Override
	protected void beforeUnLoadConfiguredDescriptor(LdapChangeLogMonitorDescriptor descriptor) {
		stopAndRemoveLdapChangeLogMonitor(descriptor);
	}

	@Override
	protected void setupDescriptorIdentifier(LdapChangeLogMonitorDescriptor descriptor, String id) {
	}

	@Override
	protected Properties beforeAddDescriptor(LdapChangeLogMonitorDescriptor descriptor) {
		Properties props = new Properties();
		props.setProperty("ldapContext", descriptor.getLdapContext());
		props.setProperty("monitorInterval", Long.toString(descriptor.getMonitorInterval()));
		props.setProperty("retrieveCount", Integer.toString(descriptor.getRetrieveCount()));
		props.setProperty("startMode", descriptor.getStartMode().name());
		props.setProperty("title", descriptor.getTitle());
		props.setProperty("description", descriptor.getDescription());
		props.setProperty("changeLogParentDN", descriptor.getChangeLogParentDN());
		props.setProperty("changeLogObjectClass", descriptor.getChangeLogObjectClass());
		props.setProperty("changeLogEntityID", descriptor.getChangeLogEntityID());
		props.setProperty("changeTypePropertyName", descriptor.getChangeTypePropertyName());
		props.setProperty("changeLogPropertyNames", descriptor.getChangeLogPropertyNames());
		props.setProperty("changeTypeMapNames", descriptor.getChangeTypeMapNames());
		props.setProperty("targetEntityPropertyName", descriptor.getTargetEntityPropertyName());
		props.setProperty("targetEntityPropertyFilter", descriptor.getTargetEntityPropertyFilter());
		props.setProperty("ordinal", Integer.valueOf(descriptor.getOrdinal()).toString());
		return props;
	}

	@Override
	protected void postAddDescriptor(LdapChangeLogMonitorDescriptor descriptor) {
		createAndStartLdapChangeLogMonitor(descriptor);
	}

	@Override
	protected void validateDescriptor(LdapChangeLogMonitorDescriptor descriptor) {
		Assert.notNull(descriptor.getLdapContext(), "必须设置数据源名称。");
		Assert.isTrue(descriptor.getMonitorInterval() > 0, "监听间隔必须大于0秒。");
		Assert.isTrue(descriptor.getRetrieveCount() > 0, "每次获取消息数据必须大于0个。");
		Assert.isTrue(getLdapContextSourceProvider().hasLdapContextSource(descriptor.getLdapContext()),
				String.format("设置的LDAP上下文(%s)不存在。", descriptor.getLdapContext()));
	}

	@Override
	protected String resolveDescriptorIdentifier(LdapChangeLogMonitorDescriptor descriptor) {
		return descriptor.getLdapContext();
	}

	@Override
	public void initContext() {
		Assert.notNull(getLdapContextSourceProvider(), "必须设置LDAP上下文提供器参数。");
		Assert.notNull(getMessageProcessor(), "必须设置消息处理器。");
		Assert.notNull(getDataTransformer(), "必须设置数据转换器。");
		Assert.notNull(getDataBroadcaster() != null || getDataReceiver() != null, "必须设置数据广播器或数据接收器。");
		super.initContext();
	}

	// implements DataSourceMonitorManager
	public void addMonitor(LdapChangeLogMonitorDescriptor info) {
		addDescriptor(info);
	}

	public void removeMonitor(String dsName) {
		removeDescriptor(dsName);
	}

	public boolean isMonitored(String dsName) {
		return hasDescriptor(dsName);
	}

	// implements DataIntegrateProcessor
	public boolean supportData(Class<? extends DataEdgeDescriptor> descriptorClazz) {
		return ClassUtils.isAssignable(LdapDataEdgeDescriptor.class, descriptorClazz);
	}

	public boolean supportData(DataEdgeDescriptor dataDescriptor) {
		return supportData((Class<? extends DataEdgeDescriptor>) ClassUtils.getUserClass(dataDescriptor));
	}

	/**
	 * @param batch
	 *            {@link DataIntegrateBatch#getInputParameter()}
	 *            必须传入一个对象数组（共两个元素）<br/>
	 *            其中{@link DataIntegrateBatch#getInputParameter()}[0]必须放置
	 *            {@link DataAction}类型。<br/>
	 *            其中{@link DataIntegrateBatch#getInputParameter()}
	 *            [1]必须放置对象数据或者列表。
	 */
	public void process(DataIntegrateBatch batch) {
		Assert.notNull(batch);
		Assert.notNull(batch.getDataDescriptor());
		Assert.notNull(batch.getInputParameter());
		Assert.isTrue(supportData(batch.getDataDescriptor()));
		try {
			List records = null;
			if (ClassUtils.isAssignableValue(List.class, batch.getInputParameter())) {
				records = (List<Object>) batch.getInputParameter();
			} else if (batch.getInputParameter().getClass().isArray()) {
				records = Arrays.asList(ObjectUtils.toObjectArray(batch.getInputParameter()));
			}
			Assert.isTrue(records.size() > 0 && records.size() <= 2, "处理参数错误(仅允许1个或两个)。");
			DataAction dataAction = DataAction.INSERT;
			if (records.size() > 1) {
				Assert.isInstanceOf(DataAction.class, records.get(0), "处理参数错误，第一个参数必须为DataAction类型。");
				dataAction = (DataAction) records.get(0);
				Object recordParameter = records.get(1);
				if (ClassUtils.isAssignableValue(List.class, recordParameter)) {
					records = (List<Object>) recordParameter;
				} else if (batch.getInputParameter().getClass().isArray()) {
					records = Arrays.asList(ObjectUtils.toObjectArray(recordParameter));
				}
			}

			Assert.notEmpty(records, "处理的记录数必须大于0。");
			Assert.isInstanceOf(Map.class, records.get(0), "记录必须为Map类型。");
			MessageEntity msgEntity = getMessageProcessor().newMessage(null,
					batch.getDataDescriptor().getMessageCode(), dataAction, records, null);

			byte[] msgBytes = ((LdapDataEdgeDescriptor) batch.getDataDescriptor()).isConversion() ? getDataTransformer()
					.transform(batch.getDataDescriptor(), msgEntity) : getMessageProcessor()
					.toMessageBytes(msgEntity, MessageFormat.XML).toString().getBytes(getMessageByteFormat());
			
			if (getDataBroadcaster() != null) {
				getDataBroadcaster().broadcast(batch.getDataDescriptor(), msgBytes);
			} 
			
			if (getDataReceiver() != null) {
				MessageEntity tmpMsg = getMessageProcessor().fromMessageBytes(msgBytes);
				getDataReceiver().receiveData(tmpMsg);
			}
		} catch (Throwable e) {
			batch.setException(new EdgeException(batch.getDataDescriptor(), e.getMessage(), e));
		}
	}

	@Override
	public Set<Class<? extends EdgeDescriptor>> getDependClasses() {
		return dependsClasses;
	}

	@Override
	public boolean dependDescriptor(EdgeDescriptor descriptor) {
		Assert.notNull(descriptor);
		return getDependClasses().contains(ClassUtils.getUserClass(descriptor.getClass())) ? isMonitored(((LdapContextDescriptor) descriptor)
				.getId()) : false;
	}

	@Override
	public LdapChangeLogMonitorDescriptor[] getDescriptorByDepend(EdgeDescriptor descriptor) {
		LdapChangeLogMonitorDescriptor returnMonitor = null;
		if (dependDescriptor(descriptor)) {
			returnMonitor = getDescriptor(((LdapContextDescriptor) descriptor).getId());
		}
		return returnMonitor == null ? new LdapChangeLogMonitorDescriptor[0]
				: new LdapChangeLogMonitorDescriptor[] { returnMonitor };
	}
	

	static class LdapChangeLogMonitorContextImpl implements LdapChangeLogMonitorContext {

		private final ContextSource contextSource;

		private final LdapChangeLogMonitorDescriptor monitorDescriptor;

		private final String systemCode;

		private final LdapDataDescriptorRegistry<LdapDataEdgeDescriptor> dataDescriptorRegistry;

		private final DataIntegrateProcessor dataIntegrateProcessor;

		private final ChangeLogResolver changeLogResolver;
		
		private final ChangeLogStorage changeLogStorage;

		LdapChangeLogMonitorContextImpl(String systemCode, ContextSource dataSource,
				ChangeLogResolver changeLogResolver, ChangeLogStorage storage, LdapChangeLogMonitorDescriptor monitorDescriptor,
				LdapDataDescriptorRegistry<LdapDataEdgeDescriptor> dataDescriptorRegsitry,
				DataIntegrateProcessor dataIntegrateProcessor) {
			this.changeLogResolver = changeLogResolver;
			this.monitorDescriptor = monitorDescriptor;
			this.contextSource = dataSource;
			this.systemCode = systemCode;
			this.dataDescriptorRegistry = dataDescriptorRegsitry;
			this.dataIntegrateProcessor = dataIntegrateProcessor;
			this.changeLogStorage = storage;
		}

		public ContextSource getLdapContextSource() {
			return contextSource;
		}

		public String getNamespace() {
			return systemCode;
		}

		public LdapDataEdgeDescriptor[] getMonitoredDataDescriptors() {
			return dataDescriptorRegistry.getDescriptors();
		}

		public LdapChangeLogMonitorDescriptor getChangeLogMonitorDescriptor() {
			return monitorDescriptor;
		}

		public ChangeLogResolver getChangeLogResolver() {
			return changeLogResolver;
		}
		
		public ChangeLogStorage getChangeLogStorage() {
			return changeLogStorage;
		}

		public void processMonitorData(LdapDataEdgeDescriptor descriptor, DataAction dataAction,
				List<Map<DataProperty, String>> changedRows) {
			Assert.notNull(descriptor);
			if (CollectionUtils.isEmpty(changedRows)) return;
			dataAction = dataAction == null ? DataAction.INSERT : dataAction;
			BaseDataIntegrateBatch batch = new BaseDataIntegrateBatch(descriptor, new Object[] { dataAction,
					changedRows });
			dataIntegrateProcessor.process(batch);
			if (batch.hasException()) {
				Throwable e = batch.getException();
				Log.error(e.getMessage(), e);
			}
		}
	}
}
